use async_trait::async_trait;
use fluviox_shared::{Result, FluvioXError, models::DataRecord};
use crate::traits::{DestinationConnector, ConnectorConfig, ConnectorStatus, ConnectorMetrics};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sqlx::PgPool;
use tracing::{info, error};
use chrono::Utc;

/// 数据库目标连接器配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabaseDestinationConfig {
    /// 数据库连接URL
    pub connection_url: String,
    /// 目标表名
    pub table_name: String,
    /// 写入模式
    pub write_mode: WriteMode,
    /// 批次大小
    pub batch_size: u32,
    /// 字段映射
    pub field_mapping: Option<std::collections::HashMap<String, String>>,
    /// 是否自动创建表
    pub auto_create_table: bool,
    /// 主键字段
    pub primary_key: Option<String>,
}

/// 写入模式
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum WriteMode {
    /// 插入模式（INSERT）
    Insert,
    /// 更新插入模式（UPSERT）
    Upsert,
    /// 替换模式（REPLACE）
    Replace,
    /// 仅更新模式（UPDATE）
    Update,
}

impl ConnectorConfig for DatabaseDestinationConfig {
    fn validate(&self) -> Result<()> {
        if self.connection_url.is_empty() {
            return Err(FluvioXError::Validation("数据库连接URL不能为空".to_string()));
        }
        
        if self.table_name.is_empty() {
            return Err(FluvioXError::Validation("表名不能为空".to_string()));
        }
        
        if self.batch_size == 0 {
            return Err(FluvioXError::Validation("批次大小必须大于0".to_string()));
        }
        
        Ok(())
    }
    
    fn to_json(&self) -> Result<Value> {
        Ok(serde_json::to_value(self)?)
    }
    
    fn from_json(value: Value) -> Result<Self> {
        Ok(serde_json::from_value(value)?)
    }
}

/// 数据库目标连接器
pub struct DatabaseDestinationConnector {
    config: Option<DatabaseDestinationConfig>,
    pool: Option<PgPool>,
    status: ConnectorStatus,
    metrics: ConnectorMetrics,
}

impl DatabaseDestinationConnector {
    pub fn new() -> Self {
        Self {
            config: None,
            pool: None,
            status: ConnectorStatus::Uninitialized,
            metrics: ConnectorMetrics::default(),
        }
    }
    
    /// 构建插入SQL
    fn build_insert_sql(&self, config: &DatabaseDestinationConfig, records: &[DataRecord]) -> Result<(String, Vec<Vec<Value>>)> {
        if records.is_empty() {
            return Err(FluvioXError::Internal("记录列表为空".to_string()));
        }
        
        // 从第一条记录获取字段列表
        let first_record = &records[0];
        let fields: Vec<String> = if let Value::Object(obj) = &first_record.data {
            obj.keys().cloned().collect()
        } else {
            return Err(FluvioXError::Internal("记录数据不是对象格式".to_string()));
        };
        
        // 应用字段映射
        let mapped_fields: Vec<String> = if let Some(mapping) = &config.field_mapping {
            fields.iter().map(|f| {
                mapping.get(f).unwrap_or(f).clone()
            }).collect()
        } else {
            fields.clone()
        };
        
        // 构建SQL
        let placeholders: Vec<String> = (1..=mapped_fields.len())
            .map(|i| format!("${}", i))
            .collect();
        
        let sql = match config.write_mode {
            WriteMode::Insert => {
                format!(
                    "INSERT INTO {} ({}) VALUES ({})",
                    config.table_name,
                    mapped_fields.join(", "),
                    placeholders.join(", ")
                )
            },
            WriteMode::Upsert => {
                if let Some(pk) = &config.primary_key {
                    let update_fields: Vec<String> = mapped_fields.iter()
                        .filter(|f| *f != pk)
                        .map(|f| format!("{} = EXCLUDED.{}", f, f))
                        .collect();
                    
                    format!(
                        "INSERT INTO {} ({}) VALUES ({}) ON CONFLICT ({}) DO UPDATE SET {}",
                        config.table_name,
                        mapped_fields.join(", "),
                        placeholders.join(", "),
                        pk,
                        update_fields.join(", ")
                    )
                } else {
                    return Err(FluvioXError::Validation("UPSERT模式需要指定主键".to_string()));
                }
            },
            _ => {
                return Err(FluvioXError::Internal("暂不支持的写入模式".to_string()));
            }
        };
        
        // 准备参数
        let mut params_batch = Vec::new();
        for record in records {
            let mut params = Vec::new();
            if let Value::Object(obj) = &record.data {
                for field in &fields {
                    let value = obj.get(field).unwrap_or(&Value::Null).clone();
                    params.push(value);
                }
            }
            params_batch.push(params);
        }
        
        Ok((sql, params_batch))
    }
}

#[async_trait]
impl DestinationConnector for DatabaseDestinationConnector {
    type Config = DatabaseDestinationConfig;
    
    fn name(&self) -> &str {
        "database-destination"
    }
    
    fn version(&self) -> &str {
        "1.0.0"
    }
    
    fn description(&self) -> &str {
        "数据库目标连接器，支持PostgreSQL等，支持多种写入模式"
    }
    
    async fn initialize(&mut self, config: Self::Config) -> Result<()> {
        info!("初始化数据库目标连接器");
        
        // 验证配置
        config.validate()?;
        
        // 创建连接池
        let pool = PgPool::connect(&config.connection_url).await
            .map_err(|e| FluvioXError::Database(e))?;
        
        self.pool = Some(pool);
        self.config = Some(config);
        self.status = ConnectorStatus::Initialized;
        
        info!("数据库目标连接器初始化完成");
        Ok(())
    }
    
    async fn test_connection(&self) -> Result<bool> {
        info!("测试数据库连接");
        
        if let Some(pool) = &self.pool {
            match sqlx::query("SELECT 1").fetch_one(pool).await {
                Ok(_) => {
                    info!("数据库连接测试成功");
                    Ok(true)
                },
                Err(e) => {
                    error!("数据库连接测试失败: {}", e);
                    Ok(false)
                }
            }
        } else {
            Ok(false)
        }
    }
    
    async fn start(&mut self) -> Result<()> {
        info!("启动数据库目标连接器");
        self.status = ConnectorStatus::Running;
        Ok(())
    }
    
    async fn stop(&mut self) -> Result<()> {
        info!("停止数据库目标连接器");
        self.status = ConnectorStatus::Stopped;
        Ok(())
    }
    
    async fn write_records(&mut self, records: Vec<DataRecord>) -> Result<usize> {
        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("连接器未初始化".to_string()))?;
        
        let pool = self.pool.as_ref()
            .ok_or_else(|| FluvioXError::Internal("数据库连接池未初始化".to_string()))?;
        
        if records.is_empty() {
            return Ok(0);
        }
        
        info!("写入 {} 条记录到数据库", records.len());
        
        // 分批处理
        let mut total_written = 0;
        let batch_size = config.batch_size as usize;
        
        for chunk in records.chunks(batch_size) {
            match self.build_insert_sql(config, chunk) {
                Ok((sql, params_batch)) => {
                    // 执行批量插入
                    let mut tx = pool.begin().await
                        .map_err(|e| FluvioXError::Database(e))?;
                    
                    for params in params_batch {
                        // 这里简化处理，实际应该使用参数化查询
                        let values: Vec<String> = params.iter().map(|v| {
                            match v {
                                Value::String(s) => format!("'{}'", s.replace("'", "''")),
                                Value::Number(n) => n.to_string(),
                                Value::Bool(b) => b.to_string(),
                                Value::Null => "NULL".to_string(),
                                _ => format!("'{}'", v.to_string().replace("'", "''")),
                            }
                        }).collect();
                        
                        let final_sql = sql.replace(&format!("$1"), &values[0]);
                        // 这里需要更复杂的参数替换逻辑
                        
                        match sqlx::query(&final_sql).execute(&mut *tx).await {
                            Ok(result) => {
                                total_written += result.rows_affected() as usize;
                            },
                            Err(e) => {
                                error!("执行SQL失败: {}", e);
                                tx.rollback().await
                                    .map_err(|e| FluvioXError::Database(e))?;
                                self.metrics.errors_count += 1;
                                return Err(FluvioXError::Database(e));
                            }
                        }
                    }
                    
                    tx.commit().await
                        .map_err(|e| FluvioXError::Database(e))?;
                },
                Err(e) => {
                    error!("构建SQL失败: {}", e);
                    self.metrics.errors_count += 1;
                    return Err(e);
                }
            }
        }
        
        // 更新指标
        self.metrics.records_processed += total_written as u64;
        self.metrics.last_activity = Some(Utc::now());
        
        info!("成功写入 {} 条记录", total_written);
        Ok(total_written)
    }
    
    fn status(&self) -> ConnectorStatus {
        self.status.clone()
    }
    
    fn metrics(&self) -> ConnectorMetrics {
        self.metrics.clone()
    }
}
